package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName: Flink05_Kafka_Flink_Kafka
 * @Description:
 * @Author: kele
 * @Date: 2021/4/12 16:07
 *
 * 从kafka获取数据
 * 1、
 *
 **/
public class Flink07_Kafka_Flink_Kafka {

    public static void main(String[] args) {

        //创建table环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        //获取kafka中的数据
        tenv.executeSql("create table IK(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int)with(" +
                            " 'connector' = 'kafka', " +
                            " 'topic' = 'senion'," +
                            " 'properties.group.id' = 'Flink05_SQL_Source_Kafka'," +
                            " 'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            " 'scan.startup.mode' = 'latest-offset', " +
                            " 'format' = 'json') ");

        tenv.executeSql("create table OK(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int)with(" +
                            " 'connector' = 'kafka', " +
                            " 'topic' = 'senion1'," +
                            " 'properties.group.id' = 'Flink05_SQL_Source_Kafka'," +
                            " 'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            " 'format' = 'json'," +
                            " 'sink.partitioner'='round-robin') ");




        tenv.executeSql("insert into OK select * from IK where id = 'sensor_1' ");

      //  tenv.executeSql("insert into OK select * from IK where id='sensor_1'");

        tenv.sqlQuery("select * from OK").execute().print();
    }

}
